package filter;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 2 * @Author: 王杰
 * 3 * @Date: 2020/11/18 12:52
 * 4
 */
public class Demo03 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //开启checkpoint
        env.enableCheckpointing(5000);

        //设置并行度为1
        env.setParallelism(1);


        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        /**
         * 设置重启策略，默认的重启策略是不停的重启
         * 固定延迟重启策略: 程序出现异常的时候，重启3次，每次延迟5秒钟重启，超过3次，程序退出
         */
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000));

        DataStreamSource<String> input = env.readTextFile("data/ip31SampleEventof1_Schema.csv");


        SingleOutputStreamOperator<String> result = input.filter(new MyFilter());

        result.print();
        result.writeAsText("data/ip31SampleEventof1_Schema2.csv");
        env.execute();
    }
}
